
//
// test_001.cpp
//     test SMPUT operation
//

#include <stdio.h>
#include <string.h>

#include <map>
#include <sstream>
#include <iostream>
#include <iomanip>

#include <boost/bind.hpp>
#include <boost/random.hpp>
#include <boost/thread/thread.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include <boost/filesystem.hpp>

#include <boost/program_options.hpp>
namespace po = boost::program_options;

using namespace boost::posix_time;

#include "fecab/client.hpp"

#define	N_ROUND	10000

typedef	boost::mt19937	random_generator_type;


float
getTimeDuration( ptime time_begin, ptime time_end )
{
    time_period tp( time_begin, time_end );
    time_duration   td = tp.length();
    long    time_last_ms = td.total_milliseconds();
    float   time_last_s = time_last_ms / 1000.0f;

    return time_last_s;
}


void
testfunc( const std::string& appconf, const std::string& logconf, int times, bool verbose, int id )
{
	char	table_name[ 32 ];

	//fecab::client::Client	clt( appconf.c_str(), logconf.c_str() );
	fecab::client::Client	clt( appconf.c_str() );

	{
		ptime   time_start = microsec_clock::local_time();
		ptime	time_start_R = time_start;
		u_int32_t	count_down_R = N_ROUND;
		u_int32_t	count_R = 0;

		u_int64_t	count_ops = 0;
		u_int32_t	count_ops_R = 0;

		std::string	script ( "passthru" );
		std::string	crumb( "crumb" );

		random_generator_type generator(42u * id);
		//boost::uniform_int<> uni_distribution(1, 65536) ;
		//boost::variate_generator<random_generator_type&, boost::uniform_int<> > rnd_int(generator, uni_distribution);
		boost::normal_distribution<> norm_dist(32768.0, 1024.0);
		boost::variate_generator<random_generator_type&, boost::normal_distribution<> > rnd(generator, norm_dist);


		typedef	std::map< int, int >	table_counter_type;
		typedef	table_counter_type::iterator	table_counter_iterator_type;
		table_counter_type	table_counter;
		table_counter_type	table_counter_R;

		for (int i=0; i<times; ++i) {
			char	key[32];
			char	value[32];

			//int	table_id = (rnd_int() % 65536) + 1;
			int		table_id = ( ((int)rnd()) % 65536 ) + 1;
			if ( table_id < 0 ) {
				table_id = -table_id;
			}
			snprintf(table_name, sizeof(table_name) - 1, "T%05u", table_id );

			std::vector< fecab::client::kv_type >	kvs_in, kvs_out;
			//int	records = (uni_int() % 64) + 1;
			int	records = ( ((int)rnd()) % 64 ) + 1;
			if (records < 0 ) {
				records =- records;
			}
			for (int k=0; k<records; ++k) {
				snprintf( key, sizeof(key) - 1, "KEY-%08d", i*1000 + k + 1 );
				snprintf( value, sizeof(value) - 1, "DATA-%08d", i*1000 + k + 1 );

				std::string	key_s( key, strlen(key) );
				std::string	value_s( value, strlen(value) );

				fecab::client::kv_type	kv;
				kv.key = key_s;
				kv.value = value_s;
				kvs_in.push_back( kv );
			}

			int rc = clt.smput( table_name, kvs_in, script, crumb, kvs_out );
			if (rc < 0) {
				std::ostringstream	msg;
				msg << "fail to SMPUT(table:" << table_name << ") i=" << i << " RC=" << rc;
				throw std::runtime_error( msg.str() );
			}

			if ( kvs_out.size() != kvs_in.size() ) {
				std::ostringstream	msg;
				msg << "SMPUT(table:" << table_name << ") -- GOT " << kvs_out.size() << " results, but WANT " << kvs_in.size();
				throw std::runtime_error( msg.str() );
			}

			for (size_t k=0; k<kvs_in.size(); ++k) {
				const fecab::client::kv_type&	kv_in  = kvs_in[ k ];
				const fecab::client::kv_type&	kv_out = kvs_out[ k ];
				if ( kv_in.key != kv_out.key ) {
					std::ostringstream	msg;
					msg << "SMPUT(table:" << table_name << ") Bad Key -- GOT:" << kv_out.key << " WANT:" << kv_in.key;
					throw std::runtime_error( msg.str() );
				}
				if ( kv_in.value != kv_out.value ) {
					std::ostringstream	msg;
					msg << "SMPUT(table:" << table_name << ", key:" << kv_in.key << ") -- GOT:" << kv_out.value << " WANT:" << kv_in.value;
					throw std::runtime_error( msg.str() );
				}
			}

			count_ops_R += records;
			count_ops += records;

			table_counter_iterator_type	it = table_counter.find( table_id );
			if ( table_counter.end() == it ) {
				table_counter.insert( std::make_pair( table_id, records ) );
			}
			else {
				it->second += records;
			}

			table_counter_iterator_type	it_r = table_counter_R.find( table_id );
			if ( table_counter_R.end() == it_r ) {
				table_counter_R.insert( std::make_pair( table_id, records ) );
			}
			else {
				it_r->second += records;
			}

			--count_down_R;
			if (0 == count_down_R) {
				ptime   time_stop_R = microsec_clock::local_time();
				float   time_last_s = getTimeDuration( time_start_R, time_stop_R );
				float   tps = N_ROUND / time_last_s;
				float	ops_tps = count_ops_R / time_last_s;
				std::cerr << "  R-SMPUT " << ++count_R << "*" << N_ROUND << " times last " << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;
				std::cerr << "          execute "  << count_ops_R << " ops on " << table_counter_R.size() << " tables, tps=" << std::setprecision(3) << std::fixed << ops_tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;

				time_start_R = time_stop_R;
				count_down_R = N_ROUND;
				count_ops_R = 0;
				table_counter_R.clear();
			}
		}

		ptime   time_stop = microsec_clock::local_time();
		float   time_last_s = getTimeDuration( time_start, time_stop );
		float   tps = times / time_last_s;
		float	ops_tps = count_ops / time_last_s;
		std::cerr << "SMPUT " << times << " times last " << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;
		std::cerr << "      execute "  << count_ops << " ops on " << table_counter.size() << " tables, tps=" << std::setprecision(3) << std::fixed << ops_tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;
	}


}


int		main(int argc, char* argv[])
{

    std::string appconf;
    std::string logconf;
    int times = 0;
    int thread_count = 0;
    bool verbose = false;

    try {

        po::options_description desc("Allowed options");
        desc.add_options()
            ("help,h", "produce help message")
            ("verbose,v", "output verbose message")
            ("app-conf", po::value<std::string>(&appconf)->default_value("app.conf"), "application config filename")
            ("log-conf", po::value<std::string>(&logconf)->default_value("log.conf"), "log config filename")
            ("times", po::value<int>(&times)->default_value(10), "number of request to send")
            ("threads", po::value<int>(&thread_count)->default_value(1), "number of threads to run")
        ;

        po::variables_map vm;
        po::store(po::parse_command_line(argc, argv, desc), vm);
        po::notify(vm);

        if (vm.count("help")) {
            std::cout << desc << "\n";
            return 1;
        }

        if (vm.count("verbose")) {
            verbose = true;
        }

    }
    catch(std::exception& e) {
        std::cerr << "error: " << e.what() << "\n";
        return 1;
    }
    catch(...) {
        std::cerr << "Exception of unknown type!\n";
        return 2;
    }


    boost::filesystem::path appconf_path( appconf );
    if ( !boost::filesystem::exists( appconf_path ) ) { 
		std::cerr << "config file " << appconf << " does NOT exist.\n";
		return 3;
    }   
    if ( !boost::filesystem::is_regular_file( appconf_path ) ) { 
		std::cerr << "config file " << appconf << " is NOT a regular file.\n";
		return 4;
    }   

	
    boost::filesystem::path logconf_path( logconf );
    if ( !boost::filesystem::exists( logconf_path ) ) { 
		std::cerr << "config file " << logconf << " does NOT exist.\n";
		return 3;
    }   
    if ( !boost::filesystem::is_regular_file( logconf_path ) ) { 
		std::cerr << "config file " << logconf << " is NOT a regular file.\n";
		return 4;
    }   
	


    std::vector< boost::thread * >    threads;

    for (int i=0; i<thread_count; i++) {
        boost::thread*   thr = new boost::thread( boost::bind(testfunc, appconf, logconf, times, verbose, i + 3) );

        threads.push_back( thr );
    }

    for (size_t i=0; i<threads.size(); i++) {
        boost::thread*  thr = threads[ i ];
        thr->join();
    }

    for (size_t i=0; i<threads.size(); i++) {
        boost::thread*  thr = threads[ i ];
        delete thr;
    }
    threads.clear();


	return	0;
}

